package com.xebia.lightning.scheduler;

import java.util.HashMap;
import java.util.Map;

import org.apache.commons.lang.StringUtils;
import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.mapreduce.MapReduceResults;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import com.xebia.lightning.converter.LocationJobCountVO;
import com.xebia.lightning.converter.LocationSkillsCountVO;
import com.xebia.lightning.domain.JobAggregate;
import com.xebia.lightning.service.JobAggregateService;
import com.xebia.lightning.service.JobService;

//@Component
public class JobAggregator {

    private Logger logger = Logger.getLogger(JobAggregator.class);

    @Autowired
    private MongoTemplate mongoTemplate;

    @Autowired
    private JobService jobService;

    @Autowired
    private JobAggregateService jobAggregateService;

    @Scheduled(fixedRate = 480000l)
    public void runJobAggregation() throws Exception {
        if (CollectionUtils.isEmpty(jobService.findAllJobs())) {
            logger.info("No job exists to run aggregation on....");
            return;
        }
        logger.info("Running Job Aggregation .......................... ");
        MapReduceResults<LocationJobCountVO> mapReduceResults = mongoTemplate
                .mapReduce("job", "classpath:map-job-aggregation.js",
                        "classpath:reduce-job-aggregation.js",
                         LocationJobCountVO.class);

        logger.info("After Map Reduce LocationJobCountVO found : "
                + mapReduceResults.getCounts().getOutputCount());

        MapReduceResults<LocationSkillsCountVO> mapReduceResultsForSkills = mongoTemplate
                .mapReduce("job", "classpath:map-per-location-skills-count.js",
                        "classpath:reduce-per-location-skills-count.js",
                        LocationSkillsCountVO.class);

        logger.info("After Map Reduce LocationSkillsCountVO found : "
                + mapReduceResultsForSkills.getCounts().getOutputCount());

        Map<String, String> map = convertToMap(mapReduceResultsForSkills);

        for (LocationJobCountVO value : mapReduceResults) {
            JobAggregate jobAggregate = convertValueObjToJobAggregate(value);
            String skillsMessage = map.get(value.getId());
            int index = StringUtils.lastIndexOf(skillsMessage, ",");
            if (index != -1) {
                skillsMessage = skillsMessage.substring(0, index);
            }
            jobAggregate.setSkillsCountMessage(skillsMessage);
            jobAggregateService.updateJobAggregate(jobAggregate);
        }
    }

    private Map<String, String> convertToMap(MapReduceResults<LocationSkillsCountVO> mapReduceResultsForSkills) {
        Map<String, String> map = new HashMap<String, String>();
        for (LocationSkillsCountVO vo : mapReduceResultsForSkills) {
            String skill = vo.getSkill();
            long count = vo.getCount();
            String location = vo.getLocation();
            String skillsMessage = new StringBuilder().append(skill).append(" : ").append(count).toString();
            String existingSkillsMessage = map.get(location);
            if (existingSkillsMessage == null) {
                map.put(location, skillsMessage);
            } else {
                String message = new StringBuilder().append(existingSkillsMessage).append(" , ").append(skillsMessage).toString();
                map.put(location, message);
            }
        }
        return map;
    }

    private JobAggregate convertValueObjToJobAggregate(LocationJobCountVO value) {
        JobAggregate jobAggregate = new JobAggregate();
        jobAggregate.setId(value.getId());
        jobAggregate.setLocation(value.getId());
        jobAggregate.setTotalJobCount(value.getValue());
        return jobAggregate;
    }

}
